package com.cloud2022.kafka.consumer;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author: create by God of SJF
 * @version: v1.0
 * @description: com.cloud2022.consumer
 * @date:2021/12/17
 */
@Component
public class ConsumerStockFactoryOut {

    @Resource
    MongoTemplate mongoTemplate;

    @KafkaListener(id = "factoryOutConsumer", topics = "stock8",concurrency = "8",containerFactory = "batchFactory")
    public void listen1(List<ConsumerRecord<String, String>> records) {
         for (ConsumerRecord<String, String> record : records) {
             String value = record.value();
             String[] values = value.split("\\|");
             value =values[0];
             String key = record.key();
             long offset = record.offset();
             System.out.println(offset);
             Stock stock = mongoTemplate.findOne(new Query(Criteria.where("barcode").is(key)), Stock.class);
             if(null != stock){
                 Stock.SdOut kafkaValue = JSONObject.parseObject(value,Stock.SdOut.class);
                 Stock.SdOut sdOut = stock.getSdOut();
                 Update update = new Update();
                 if(null != sdOut){
                      if(sdOut.getSdOutTime().before(kafkaValue.getSdOutTime())){
                          update.set("sdOut",kafkaValue);
                      }else {
                          continue;
                      }
                 }else{
                     update.set("sdOut",kafkaValue);
                 }
                 mongoTemplate.updateFirst(new Query(Criteria.where("barcode").is(key)),update,Stock.class);

             }
         }
    }
}
